[SPARK-61097][CONNECT][PYTHON] Cancel listener work during session shutdown - #384

What changes were proposed in this pull request?

Fix shutdown of the shared streaming listener bus. For example, external streams.close() can wait for a callback while the callback waits to remove itself. Lifecycle RPCs and worker joins now run outside the registry lock. Ordinary failed removal or close preserves retry state.

Terminal stop cancels local registration/dispatch, releases the server session, then closes the client even if release fails. Client close cancels locally without releasing the server session. Workers hold weak bus references, and progress tracking shares its handler registry while retaining recursion limits.

Preserve upstream duplicate-registration, foreign-session removal and synchronous-start behavior. Consume one Python registration ID only after completed removal, and protect newer registrations from older worker generations.

Why are the changes needed?

SPARK-61097. Callback shutdown can deadlock, failed close can discard retry state, and a quiet worker can retain session ownership.

Includes the NumPy prerequisite (SPARK-61102).

Does this PR introduce any user-facing change?

Yes. Listener teardown avoids callback waits, ordinary removal failures remain retryable, and terminal teardown cancels quiet dispatch. A current event snapshot may finish afterward. Acknowledgement cancellation begins after stream establishment; initial establishment retains transport deadlines/retries. Clearing registrations after clean stream end is a robustness behavior, not established official parity. Listener-held session cycles and event-error warning parity remain limitations.

How was this patch tested?

On final source ad40f31:

• 617 Rust library units and 2,203 Python offline/Arrow-handoff tests passed; eight live-server tests skipped. Normal and WASM targets compiled 42 executables each; compilation did not execute them.
• Lifecycle stress: 25 rounds of 27 mock-server cases, 675 passing executions, no failures/errors/skips. Each round deselected 1,209 unrelated cases.
• Nine native live cases and all 27 phase reports passed with imports verified; 180 unrelated cases were deselected. Three exact Rust real-worker listener tests passed. The ownership control passed, while a strong-reference mutant failed at the final listener-drop assertion. An earlier mutation attempt stopped during cleanup before executing the mutant and remains recorded.
• Scoped official streaming/listener/progress checks: 48 passed (42 substantive cases and six helpers), no skips or retries. API inspection found no missing inspected members across 12 surfaces. These checks do not establish complete signatures or native high-level parity; Row tuple compatibility is separate.
• This branch's own fresh transport binary completed 87 normal full-suite files: 2,586 passed, 641 skipped, one deselected, no failures/errors/retries. Eleven files were entirely skipped. Its separate four-module cluster phase passed 15 invocations with 14 JVM-guarded skips and no retries. The NumPy prerequisite independently completed the same full/cluster totals.

The unchanged metadata-column manifest entry still failed a separate advisory check because its test catalog plugin was unavailable (one failure, 57 deselections). The original NumPy full run's seeded-random failure remains recorded: the unchanged test failed identically across official Python and both saved Rust transports on two threads, then passed in all three on its default four-thread fixture, used by the complete reruns.

Full/cluster checks use official Python plans/converters over the respective native raw transports, without per-child import attestation. They do not validate native high-level Row/literal/config/listener behavior, native ReleaseSession or native artifact upload. The recursion subprocess also lacks separate import telemetry. Logs retain deliberate worker crashes, negative tests, in-test query restarts, cursor errors and shutdown races. Direct successful server shutdown was recorded without universal descendant-process telemetry.

Clippy remains advisory exit 101 with 144 lint errors in unchanged source regions. A fresh baseline/equal-coverage comparison is unavailable. Linux, Windows and the newest-master combined tree were not runtime tested locally.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex

This pull request and its description were written by OpenAI Codex.